package com.gzyj.flink.gps.map.abnormaloffline;

import cn.hutool.core.lang.Pair;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.fastjson.support.spring.FastJsonRedisSerializer;
import com.gzyj.flink.driver.Driverlogininfo;
import com.gzyj.flink.dto.alarm.AbnormalOfflineData;
import com.gzyj.flink.dto.alarm.Kafka_VehicleRemindContent;
import com.gzyj.flink.gps.GPSPosition;
import com.gzyj.flink.gps.config.RedisKeyPrefixConfig;
import com.gzyj.flink.gps.factory.MySQLMybatisSessionFactory;
import com.gzyj.flink.gps.factory.RedisSessionFactory;
import com.gzyj.flink.gps.map.fatiguedriving.FatigueDrivingResult;
import com.gzyj.flink.gps.mapper.mysql.MonYichangSetMapper;
import com.gzyj.flink.gps.mapper.mysql.TsAlarmRuleMapper;
import com.gzyj.flink.gps.service.NormalAlarmService;
import com.gzyj.flink.gps.utils.IntUtils;
import com.gzyj.flink.gps.utils.TimeUtils;
import com.gzyj.flink.pojo.mysql.set.MonYichangSet;
import com.gzyj.flink.pojo.mysql.set.TsAlarmRule;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import java.util.Date;
import java.util.concurrent.TimeUnit;

public class AbNormalofflineDataProcess extends RichFlatMapFunction<GPSPosition, FatigueDrivingResult> {
    LettuceConnectionFactory redisFactory;

    SqlSessionFactory mysql;

    NormalAlarmService normalAlarmService;

    private RedisTemplate<String, Object> getRedisTemplate(LettuceConnectionFactory redisSessionFactory) {
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(redisSessionFactory);
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(new FastJsonRedisSerializer<>(Object.class));
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashValueSerializer(new FastJsonRedisSerializer<>(Object.class));
        redisTemplate.afterPropertiesSet();
        return redisTemplate;
    }


    @Override
    public void open(Configuration parameters) throws Exception {
        redisFactory = RedisSessionFactory.getRedisFactory();
        mysql = MySQLMybatisSessionFactory.getSqlSessionFactory();
        normalAlarmService = new NormalAlarmService();
    }

    private AbnormalOfflineData TryGetData_ABD(String vehId, RedisTemplate<String, Object> clusterRedisTemplate) {

        Object obj = clusterRedisTemplate.opsForValue().get(RedisKeyPrefixConfig.RedisKey_AbnormalOffline + vehId);

        if (obj == null) {
            return new AbnormalOfflineData();
        } else {
            return JSON.parseObject(JSON.toJSONString(obj, SerializerFeature.DisableCircularReferenceDetect), AbnormalOfflineData.class);
        }


    }

    public Pair<Boolean, MonYichangSet> GetRuleSetting(GPSPosition vehicle, MonYichangSetMapper monYichangSetMapper, TsAlarmRuleMapper tsAlarmRuleMapper) {
        MonYichangSet setting = getVehicleForAbnormalOfflineParamsDic(vehicle.getVehId(), monYichangSetMapper);
        if (setting != null) {
            return new Pair<>(true, setting);
        } else {
            TsAlarmRule rule = tsAlarmRuleMapper.selectByRuleTypeAbnormal(vehicle.getShiyongxingzhi());
            if (rule == null || rule.getRuleFlag() == 0) {
                return new Pair<>(false, null);
            } else {
                setting = getVehicleForAbnormalOfflineParamsDic(rule.getRuleNo(), monYichangSetMapper);
                if (setting != null) {
                    return new Pair<>(true, setting);
                } else {
                    return new Pair<>(false, null);
                }
            }


        }
    }

    private MonYichangSet getVehicleForAbnormalOfflineParamsDic(String vehId, MonYichangSetMapper monYichangSetMapper) {
        return monYichangSetMapper.selectByVehID(vehId);
    }

    public Driverlogininfo TryGetData_DLD(String vehId, RedisTemplate<String, Object> clusterRedisTemplate) {
        Object obj = clusterRedisTemplate.opsForValue().get(RedisKeyPrefixConfig.RedisKey_DriverSignOn + vehId);
        if (obj == null) {
            return new Driverlogininfo();
        } else {
            return JSON.parseObject(JSON.toJSONString(obj, SerializerFeature.DisableCircularReferenceDetect), Driverlogininfo.class);
        }

    }

    private Kafka_VehicleRemindContent DisposeMessage(AbnormalOfflineData aboData, MonYichangSet mySetting, GPSPosition gps, String msgType) {
        return null;
    }

    @Override
    public void flatMap(GPSPosition gpsPosition, Collector<FatigueDrivingResult> collector) throws Exception {

        RedisTemplate<String, Object> clusterRedisTemplate = getRedisTemplate(redisFactory);
        SqlSession sqlSession = mysql.openSession(true);
        try {

            MonYichangSetMapper monYichangSetMapper = sqlSession.getMapper(MonYichangSetMapper.class);
            TsAlarmRuleMapper tsAlarmRuleMapper = sqlSession.getMapper(TsAlarmRuleMapper.class);

            Driverlogininfo driverLoginData = TryGetData_DLD(gpsPosition.getVehId(), clusterRedisTemplate);

            AbnormalOfflineData aboData = TryGetData_ABD(gpsPosition.getVehId(), clusterRedisTemplate);

            if (aboData == null) return;
            if (StringUtils.isEmpty(gpsPosition.getTIME())) {
                return;
            }
            Pair<Boolean, Integer> p = TimeUtils.CheckData(aboData.getGpsTime(), gpsPosition.getTIME());
            int i = p.getValue();

            if (!p.getKey()) {
                return;
            } else {
                aboData.LocalTime = new Date();
                aboData.GpsTime = gpsPosition.getGPSTimeExt();
            }
            //获取运算参数
            Pair<Boolean, MonYichangSet> p1 = GetRuleSetting(gpsPosition, monYichangSetMapper, tsAlarmRuleMapper);
            MonYichangSet mySetting = p1.getValue();

            if (p1.getKey()) {
                //如果车辆配置了规则
                if (mySetting.getEnable() == 1) {
                    if (IntUtils.AccExt(gpsPosition.getSTATUSBIT()) == 1 && gpsPosition.getSpeedExt() > mySetting.getSpeed()) {
                        //终端开启acc且速度大于设定阈值
                        if (!aboData.IsCalc) {
                            aboData.ResetData();
                            aboData.StartCalc(gpsPosition);
                        }
                        aboData.WriteCalc(gpsPosition);
                        if (aboData.CalcConditionTimeSpan() >=mySetting.getOfflineTime() * 60) {
                            if (!aboData.IsAlarm) {
                                aboData.StartAlarm(gpsPosition);
                                //从没报警切换到报警，发送开始报警
                                collector.collect(new FatigueDrivingResult(normalAlarmService.SendAlarmToKafka(aboData, gpsPosition, driverLoginData, true), null));
                            }
                            //其他的时候记录报警的过程数据
                            aboData.WriteAlarm(gpsPosition);

                            //停止报警
                            if (aboData.IsAlarm) {
                                aboData.StopAlarm(gpsPosition);
                                //从报警切换到没报警，发送停止报警
                                collector.collect(new FatigueDrivingResult(normalAlarmService.SendAlarmToKafka(aboData, gpsPosition, driverLoginData, false), DisposeMessage(aboData, mySetting, gpsPosition, "报警短信")));
                            }
                            aboData.GenerateSingleGuid();
                            aboData.ResetData();
                        } else {
                            aboData.ResetData();
                            aboData.StartCalc(gpsPosition);
                            aboData.WriteCalc(gpsPosition);
                        }
                    } else {
                        //如果车辆规则关闭
                        if (aboData.IsAlarm) {
                            aboData.StopAlarm(gpsPosition);
                            //从报警切换到没报警，发送停止报警
                            collector.collect(new FatigueDrivingResult(normalAlarmService.SendAlarmToKafka( aboData, gpsPosition,driverLoginData, false),null));
                        }
                        aboData.GenerateSingleGuid();
                        aboData.ResetData();
                    }
                } else {
                    //如果车辆规则关闭
                    if (aboData.IsAlarm) {
                        aboData.StopAlarm(gpsPosition);
                        //从报警切换到没报警，发送停止报警
                        collector.collect(new FatigueDrivingResult(normalAlarmService.SendAlarmToKafka( aboData, gpsPosition,driverLoginData, false),null));
                    }
                    aboData.GenerateSingleGuid();
                    aboData.ResetData();
                }
            } else {
                //如果车辆没有配置规则或者取消了规则
                if (aboData.IsAlarm) {
                    aboData.StopAlarm(gpsPosition);
                    //从报警切换到没报警，发送停止报警
                    collector.collect(new FatigueDrivingResult(normalAlarmService.SendAlarmToKafka( aboData, gpsPosition,driverLoginData, false),null));
                }
                aboData.GenerateSingleGuid();
                aboData.ResetData();
            }
            clusterRedisTemplate.opsForValue().set(RedisKeyPrefixConfig.RedisKey_AbnormalOffline + gpsPosition.getVehId(), aboData, 8, TimeUnit.HOURS);


        } catch (Exception e) {
            e.printStackTrace();
            sqlSession.rollback();
        } finally {
            sqlSession.close();
        }


    }
}
